# -*- coding: utf-8 -*-
from pyspark import SparkConf, SparkContext

if __name__ == "__main__":
    conf = SparkConf().setAppName("PythonSparkDemo5").setMaster("yarn-client")
    sc = SparkContext(conf=conf)
    data1 = [1, 2, 3, 4, 5, 6, 6]
    data2 = [9, 8, 7, 6, 5, 4, 3]
    rdd1 = sc.parallelize(data1)
    rdd2 = sc.parallelize(data2)

    #*------------这里是对单个RDD的转化操作-------------*/
    afterMap = rdd1.map(lambda x:  x + 1)
    print("afterMap---{}".format(afterMap.collect())) # afterMap ---[2, 3, 4, 5, 6, 7, 7]

    afterFlatMap = rdd1.flatMap(lambda x: [x - 1])
    print("afterFlatMap---{}".format(afterFlatMap.collect())) # afterFlatMap---[0, 1, 2, 3, 4, 5, 5]  直接x-1运行出错，错误提示是int不支持的迭代

    afterFilter = rdd1.filter(lambda x: x > 2)
    print("afterFilter---{}".format(afterFilter.collect()))# afterFilter - --[3, 4, 5, 6, 6]

    afterDisTinct = rdd1.distinct()
    print("afterDisTinct---{}".format(afterDisTinct.collect())) # afterDisTinct---[4, 6, 2, 1, 3, 5]

    #/ *------------这里是对两个RDD的转化操作 - ------------ * /
    afterUnion = rdd1.union(rdd2)
    print("afterUnion---{}".format(afterUnion.collect())) #afterUnion---[1, 2, 3, 4, 5, 6, 6, 9, 8, 7, 6, 5, 4, 3]

    afterIntersection = rdd1.intersection(rdd2)
    print("afterIntersection---{}".format(afterIntersection.collect())) #afterIntersection---[4, 6, 3, 5]

    afterSubtarct = rdd1.subtract(rdd2)
    print("afterSubtarct---{}".format(afterSubtarct.collect())) #afterSubtarct---[2, 1]

    #/*------------这里是对RDD的行为(action)操作----------*/
    print("collect--{}".format(rdd1.collect()))# [1, 2, 3, 4, 5, 6, 6]


    print("count---{}".format(rdd1.count())) # 7

    print("countByValue---{}".format(rdd1.countByValue)) # {5 = 1, 1 = 1, 6 = 2, 2 = 1, 3 = 1, 4 = 1}

    print("take---{}".format(rdd1.take(1))) # [1]

    print("top---{}".format(rdd1.top(2))) # [6, 6]

    print("takeOrdered---{}".format(rdd1.takeOrdered(4))) # [1, 2, 3, 4]

    print("takeSample---{}".format(rdd1.takeSample(None, 1))) # [6]

    print("reduce---{}".format(rdd1.reduce(lambda x , y : x + y))) # 27

    print("fold---{}".format(rdd1.fold(1, lambda x, y : x + y))) # 30


    sc.stop()

